package bus

import (
	"context"
	"encoding/json"

	"gitee.com/go-course/go8/projects/devcloud/maudit/apps/operate"
	"github.com/infraboard/mcube/app"
	"github.com/infraboard/mcube/logger"
	"github.com/infraboard/mcube/logger/zap"
	"github.com/segmentio/kafka-go"
)

const (
	OPERATE_LOG_TOPIC = "log.operate"
)

func NewPublisher(addr []string) *Publisher {
	w := kafka.NewWriter(kafka.WriterConfig{
		Brokers: addr,
		Topic:   OPERATE_LOG_TOPIC,
	})
	return &Publisher{
		log:      zap.L().Named("bus.operate"),
		producer: w,
	}
}

// 发布者
type Publisher struct {
	log      logger.Logger
	producer *kafka.Writer
}

// 往总线上发布数据
func (b *Publisher) Pub(ctx context.Context, record *operate.OperateLog) error {
	// 1.序列化
	paylaod, err := json.Marshal(record)
	if err != nil {
		return err
	}

	// 2. 转换为kafka Message
	m := kafka.Message{
		Value: paylaod,
	}

	// 3. 发送数据
	return b.producer.WriteMessages(ctx, m)
}

func NewSubscriber(addr []string) *Subscriber {
	r := kafka.NewReader(kafka.ReaderConfig{
		Brokers: addr,
		GroupID: "consumer-group-id",
		Topic:   OPERATE_LOG_TOPIC,
	})
	return &Subscriber{
		log:      zap.L().Named("bus.operate"),
		consumer: r,
		operate:  app.GetInternalApp(operate.AppName).(operate.Service),
	}
}

// 订阅者
// 应该放在那里启动, 服务启动的时候就启动
type Subscriber struct {
	log      logger.Logger
	consumer *kafka.Reader
	operate  operate.Service
}

// 订阅总线数据
func (b *Subscriber) Sub(ctx context.Context) {
	b.log.Infof("subscribe audit bus ...")

	for {
		// 1. 获取数据, 有可能io.EOF
		m, err := b.consumer.ReadMessage(ctx)
		if err != nil {
			b.log.Errorf("read message error, %s", err)
			return
		}

		b.log.Debugf("message at topic/partition/offset %v/%v/%v: %s = %s\n", m.Topic, m.Partition, m.Offset, string(m.Key), string(m.Value))

		// 2. 反序列化
		record := operate.NewOperateLog()
		if err := json.Unmarshal(m.Value, record); err != nil {
			b.log.Errorf("unmarshal error, %s", err)
			continue
		}

		// 3. 入库到数据
		_, err = b.operate.SaveOperateLog(ctx, record)
		if err != nil {
			b.log.Errorf("save operate log error, %s", err)
			return
		}

		// 处理完消息后需要提交该消息已经消费完成, 消费者挂掉后保存消息消费的状态
		if err := b.consumer.CommitMessages(context.Background(), m); err != nil {
			b.log.Errorf("failed to commit messages:", err)
		}
	}
}
